package me.tzk.pdi.plugins.trans.steps.rocketmq;

import com.google.common.collect.ImmutableMap;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.pentaho.di.core.logging.LogChannelInterface;
import org.pentaho.di.core.util.StringUtil;
import org.pentaho.di.trans.step.StepInterface;

import java.util.List;
import java.util.Map;

public class RocketMQClientBuilder {
    private static final Class<?> PKG = RocketMQClientBuilder.class;

    protected DefaultMQPushConsumer client;

    public static final Map<String, String> DEFAULT_SSL_OPTS = ImmutableMap.<String, String>builder()
            .put("ssl.protocol", "TLS")
            .put("ssl.contextProvider", "")
            .put("ssl.keyStore", "")
            .put("ssl.keyStorePassword", "")
            .put("ssl.keyStoreType", "JKS")
            .put("ssl.keyStoreProvider", "")
            .put("ssl.trustStore", "")
            .put("ssl.trustStorePassword", "")
            .put("ssl.trustStoreType", "")
            .put("ssl.trustStoreProvider", "")
            .put("ssl.enabledCipherSuites", "")
            .put("ssl.keyManager", "")
            .put("ssl.trustManager", "")
            .build();

    private String onsAddr;
    private String customerId;
    private String topic;
    private List<String> tags;
    private String accessKey;
    private String secretKey;
    private boolean isSecure = false;
    private Map<String, String> sslConfig;
    private String messageModel;
    private String consumeThreadNums;
    private String maxReconsumeTimes;
    private String consumeTimeout;
    private String consumeMessageBatchMaxSize;
    private String checkImmunityTimeInSeconds;
    private String shardingKey;
    private String suspendTimeMillis;
    private MessageListenerConcurrently callback;
    private LogChannelInterface logChannel;
    private String stepName;
    private String consumeFromWhere;

    private RocketMQClientBuilder() {
    }

    public static RocketMQClientBuilder build() {
        return new RocketMQClientBuilder();
    }

    RocketMQClientBuilder withCallback(MessageListenerConcurrently callback) {
        this.callback = callback;
        return this;
    }

    public RocketMQClientBuilder withONSAddr(String onsAddr) {
        this.onsAddr = onsAddr;
        return this;
    }

    public RocketMQClientBuilder withCustomerId(String customerId) {
        this.customerId = customerId;
        return this;
    }

    RocketMQClientBuilder withTopic(String topic) {
        this.topic = topic;
        return this;
    }

    RocketMQClientBuilder withTags(List<String> tags) {
        this.tags = tags;
        return this;
    }

    public RocketMQClientBuilder withAccessKey(String accessKey) {
        this.accessKey = accessKey;
        return this;
    }

    public RocketMQClientBuilder withSecretKey(String secretKey) {
        this.secretKey = secretKey;
        return this;
    }

    public RocketMQClientBuilder withStep(StepInterface step) {
        this.logChannel = step.getLogChannel();
        this.stepName = step.getStepMeta().getName();
        return this;
    }

    public RocketMQClientBuilder withIsSecure(boolean isSecure) {
        this.isSecure = isSecure;
        return this;
    }

    RocketMQClientBuilder withSslConfig(Map<String, String> sslConfig) {
        this.sslConfig = sslConfig;
        return this;
    }

    public RocketMQClientBuilder withMessageModel(String messageModel) {
        this.messageModel = messageModel;
        return this;
    }

    public RocketMQClientBuilder withConsumeThreadNums(String consumeThreadNums) {
        this.consumeThreadNums = consumeThreadNums;
        return this;
    }

    public RocketMQClientBuilder withMaxReconsumeTimes(String maxReconsumeTimes) {
        this.maxReconsumeTimes = maxReconsumeTimes;
        return this;
    }

    public RocketMQClientBuilder withConsumeTimeout(String consumeTimeout) {
        this.consumeTimeout = consumeTimeout;
        return this;
    }

    public RocketMQClientBuilder withConsumeMessageBatchMaxSize(String consumeMessageBatchMaxSize) {
        this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize;
        return this;
    }

    public RocketMQClientBuilder withCheckImmunityTimeInSeconds(String checkImmunityTimeInSeconds) {
        this.checkImmunityTimeInSeconds = checkImmunityTimeInSeconds;
        return this;
    }

    public RocketMQClientBuilder withShardingKey(String shardingKey) {
        this.shardingKey = shardingKey;
        return this;
    }

    public RocketMQClientBuilder withSuspendTimeMillis(String suspendTimeMillis) {
        this.suspendTimeMillis = suspendTimeMillis;
        return this;
    }

    public RocketMQClientBuilder withConsumeFromWhere(String consumeFromWhere) {
        this.consumeFromWhere = consumeFromWhere;
        return this;
    }

    public DefaultMQPushConsumer createConsumer() throws MQClientException {
        validateArgs();

        if (StringUtil.isEmpty(customerId)) {
            logChannel.logDebug("Customer ID cannot be null.");
        }
        // 创建消费者
        client = new DefaultMQPushConsumer(customerId);
        client.setNamesrvAddr(onsAddr);
        client.subscribe(topic, buildTag(tags));
        try {
            ConsumeFromWhere fromWhere = ConsumeFromWhere.valueOf(consumeFromWhere);
            client.setConsumeFromWhere(fromWhere);
        } catch (Exception e) {
        }
        client.registerMessageListener(callback);
        // 消费线程1个，防止背压队列溢出异常
        int ctn = 1;
        try {
            ctn = Integer.parseInt(consumeThreadNums);
        } catch (NumberFormatException e) {
        }
        client.setConsumeThreadMin(ctn);
        client.setConsumeThreadMax(ctn);
        // 队列最大500，控制内存
        client.setPullThresholdForQueue(500);
        MessageModel model = MessageModel.CLUSTERING;
        try {
            model = MessageModel.valueOf(messageModel);
        } catch (Exception e) {
        }
        client.setMessageModel(model);

        logChannel.logDebug("Subscribing to topic: " + topic + " with tags: " + buildTag(tags) + " , model: " +
                model.getModeCN() + " , consumeThreadNums: " + ctn);
        return client;
    }

    private void validateArgs() {
        // check any conditions
    }

    private String buildTag(List<String> tags) {
        for (String aTag : tags) {
            if ("*".equals(aTag)) {
                return "*";
            }
        }
        return String.join("||", tags);
    }
}
